package com.sl.core.engine.rowset;

import cn.hutool.core.codec.Base64;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.exceptions.ExceptionUtil;
import cn.hutool.core.util.ArrayUtil;
import cn.hutool.core.util.BooleanUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.google.common.collect.Maps;
import com.sl.core.engine.meta.row.ProcessRowDescriptor;
import com.sl.core.engine.meta.valuemeta.ValueMetaConvert;
import com.sl.core.engine.trans.ProcessInstance;
import com.sl.core.engine.trans.ProcessTrans;
import com.zaxxer.hikari.HikariDataSource;
import lombok.SneakyThrows;
import org.anyline.data.jdbc.runtime.JDBCRuntimeHolder;
import org.anyline.data.jdbc.util.DataSourceUtil;
import org.anyline.data.runtime.RuntimeHolder;
import org.anyline.entity.DataRow;
import org.anyline.entity.DataSet;
import org.anyline.metadata.Column;
import org.anyline.metadata.Table;
import org.anyline.metadata.type.init.StandardTypeMetadata;
import org.anyline.proxy.ServiceProxy;
import org.anyline.service.AnylineService;
import org.apache.commons.compress.utils.Lists;
import org.apache.hop.core.BlockingBatchingRowSet;
import org.apache.hop.core.Const;
import org.apache.hop.core.row.IRowMeta;
import org.apache.hop.core.row.IValueMeta;

import javax.sql.DataSource;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

public abstract class ProcessAbstractDbRowSet extends BlockingBatchingRowSet {

    protected ProcessInstance processInstance;


    protected AnylineService anylineService;

    private String tid;


    private static final String TABLE_SPLIT_SPEATOR = "#_#";
    private static final Map<Integer, StandardTypeMetadata> TYPE_MAPPING = Maps.newConcurrentMap();

    AtomicBoolean ready = new AtomicBoolean(false);

    public ProcessAbstractDbRowSet(ProcessTrans processTrans) {
        super(5000);
        if (ObjectUtil.isNull(processTrans)) {
            return;
        }
        if (!(processTrans instanceof ProcessInstance processInstance)) {
            return;
        }
        this.processInstance = processInstance;
        this.tid = processInstance.getTid();
        initDb();
    }


    protected Boolean writeDbFlag() {
        return true;
    }

    protected DataSource getDataSource(String jdbcUrl, String jdbcDriver, String jdbcUserName, String jdbcPassword) {
        Map<String, Object> maps = Maps.newConcurrentMap();
        maps.put("url", jdbcUrl);
        maps.put("driver", jdbcDriver);
        maps.put("username", jdbcUserName);
        maps.put("password", jdbcPassword);
        maps.put("type", HikariDataSource.class.getName());
        DataSource dataSource = DataSourceUtil.build(maps);
        return dataSource;
    }

    protected void initDb(String jdbcUrl, String jdbcDriver, String jdbcUserName, String jdbcPassword) {

        if (ObjectUtil.isNotNull(anylineService)) {
            return;
        }

        synchronized (this) {
            if (ObjectUtil.isNotNull(anylineService)) {
                return;
            }
            DataSource dataSource = getDataSource(jdbcUrl, jdbcDriver, jdbcUserName, jdbcPassword);

            String encode = Base64.encode(jdbcUrl);
            if (!RuntimeHolder.contains(encode)) {
                JDBCRuntimeHolder.instance().reg(encode, dataSource);
            }
            anylineService = ServiceProxy.service(encode);
        }
    }

    protected List<Map<String, Object>> getStepList() {
        if (ObjectUtil.isNull(anylineService)) {
            return Collections.emptyList();
        }
        List<Map<String, Object>> result = Lists.newArrayList();
        LinkedHashMap<String, Table> tables = anylineService.metadata().tables();
        for (Map.Entry<String, Table> st : tables.entrySet()) {
            String key = st.getKey();
            Table value = st.getValue();
            if (StrUtil.equalsAnyIgnoreCase(key, "SQLITE_SEQUENCE")) {
                continue;
            }
            if (!StrUtil.containsAnyIgnoreCase(key, TABLE_SPLIT_SPEATOR)) {
                continue;
            }
            List<String> split = StrUtil.split(key, TABLE_SPLIT_SPEATOR);

            Map<String, Object> map = Maps.newHashMap();
            map.put("type", typeCode());
            map.put("source", CollUtil.get(split, 0));
            map.put("target", CollUtil.get(split, 1));
            result.add(map);
        }
        return result;
    }

    protected Map<String, Object> getStepDataList() {
        DataSet querys = anylineService.querys(getTableName(), "1=1");
//        List<DataRow> list = querys.stream().toList();
        Map<String, Object> result = Maps.newHashMap();
        result.put("columns", querys.getMetadatas());
        result.put("type", typeCode());
        result.put("data", querys.getRows());
        return result;
    }

    @SneakyThrows
    protected void initDb() {
        if (!writeDbFlag()) {
            return;
        }
        initDb(getJdbcUrl(), getJdbcDriver(), getJdbcUserName(), getJdbcPassword());
    }

    protected String getTableName() {
        String originTransformName = this.getOriginTransformName();
        String destinationTransformName = this.getDestinationTransformName();

        String tableName = originTransformName + TABLE_SPLIT_SPEATOR + destinationTransformName;
        return tableName;
    }


    @SneakyThrows
    protected void createTable(List<IValueMeta> valueMetaList) {
        initDb();
        //判断表是否存在，不存在创建，存在则不创建，但要清空数据
//        ConfigStore rootConfig = new DefaultConfigStore();
//        rootConfig.datasource()

        Table table = new Table(getTableName());

        boolean exists = anylineService.metadata().exists(table);
        //存在就不管了
        if (BooleanUtil.isTrue(exists)) {
            ready.set(true);
            return;
        }
        Column column = new Column("id");
        column.setTypeMetadata(StandardTypeMetadata.BIGINT);
        column.autoIncrement(true);
        column.setPrimaryKey(true);
        table.addColumn(column);

        customColumns(table);

        column = new Column("tid");
        column.setTypeMetadata(StandardTypeMetadata.VARCHAR);
        column.setLength(250);
        table.addColumn(column);

        column = new Column("startTime");
        column.setTypeMetadata(StandardTypeMetadata.DATETIME);
        table.addColumn(column);

        column = new Column("endTime");
        column.setTypeMetadata(StandardTypeMetadata.DATETIME);
        table.addColumn(column);
//
//
//        column = new Column("row_key");
//        column.setTypeMetadata(StandardTypeMetadata.VARCHAR);
//        table.addColumn(column);
//
//        column = new Column("row_value");
//        column.setTypeMetadata(StandardTypeMetadata.VARCHAR);
//        table.addColumn(column);


//        column = new Column("value_is_json");
//        column.setTypeMetadata(StandardTypeMetadata.INTEGER);
//        table.addColumn(column);

        for (IValueMeta iValueMeta : valueMetaList) {
            String name = iValueMeta.getName();
            int type = iValueMeta.getType();
            column = new Column(name);

            StandardTypeMetadata standardTypeMetadata = TYPE_MAPPING.get(type);
            if (ObjectUtil.isNull(standardTypeMetadata)) {
                //暂时定字符串， 后续做转换
                standardTypeMetadata = StandardTypeMetadata.VARCHAR;
            }
            column.setTypeMetadata(standardTypeMetadata);
            column.setLength(255);
            table.addColumn(column);
        }

        column = new Column("rowReadFlag");
        column.setTypeMetadata(StandardTypeMetadata.INTEGER);
        column.setLength(1);
        table.addColumn(column);

        column = new Column("rowProcessErrorFlag");
        column.setTypeMetadata(StandardTypeMetadata.INTEGER);
        column.setLength(1);
        table.addColumn(column);

        column = new Column("rowProcessErrorMsg");
        column.setTypeMetadata(StandardTypeMetadata.TEXT);
        table.addColumn(column);

        //没有就创建
        anylineService.ddl().create(table);

        processInstance.getLogChannel().logBasic(table.ddl());

        ready.set(true);

    }


    protected void customColumns(Table table) {

    }

    abstract public String typeCode();

    public List<Map<String, Object>> searchStepList(String date, String processId, String processVersion, String tid, String dbFile) {
        return Collections.emptyList();
    }

    public List<Map<String, Object>> searchProcessList(List<String> dateList) {
        return Collections.emptyList();
    }


    public Map<String, Object> searchStepDataList(String date, String processId, String processVersion, String tid, String dbFile,
                                                  String source,
                                                  String target) {
        return null;
    }

    abstract protected String getJdbcUserName();

    abstract protected String getJdbcPassword();

    abstract protected String getJdbcUrl();

    abstract protected String getJdbcDriver();


    @Override
    public Object[] getRow() {
        return getRowWait(Const.TIMEOUT_GET_MILLIS, TimeUnit.MILLISECONDS);
    }

    @Override
    public Object[] getRowImmediate() {
        return getRow();
    }


    @Override
    public Object[] getRowWait(long timeout, TimeUnit tu) {
//        Object[] rowWait = super.getRowWait(timeout, tu);
        initDb();

        if (!ready.get()) {
            return null;
        }

        String tableName = getTableName();
        DataRow query = null;

        synchronized (tableName) {
            String whereCondition = getWhereCondition();

            query = anylineService.query(tableName, StrUtil.nullToDefault(whereCondition, " 1=1 ") + " and rowReadFlag = 0 and tid = '"+this.tid+"'");
            if (ObjectUtil.isNull(query)) {
                return null;
            }
            DataRow row = new DataRow();
            row.put("ID", query.getId());
            row.put("rowReadFlag", 1);
            row.set("endTime", DateUtil.date());
            anylineService.update(tableName, row);
        }

        List<Object> objects = dbDataToRow(query);
        return objects.toArray();
    }

    protected String getWhereCondition() {

        return null;
    }

    public static Boolean isHIndex(Object value) {
        return ObjectUtil.isNotNull(value) && StrUtil.startWith(value.toString(), "hIndex_");
    }

    @Override
    public boolean putRow(IRowMeta rowMeta, Object[] rowData) {
        return putRowWait(rowMeta, rowData, Const.TIMEOUT_PUT_MILLIS, TimeUnit.MILLISECONDS);
    }

    @SneakyThrows
    @Override
    public boolean putRowWait(IRowMeta rowMeta, Object[] rowData, long time, TimeUnit tu) {

//        super.putRowWait(rowMeta, rowData, time, tu);
        super.setRowMeta(rowMeta);

        createTable(rowMeta.getValueMetaList());

        DataRow dataRow = new DataRow();
        dataRow.put("startTime", DateUtil.date());
        dataRow.put("rowReadFlag", 0);
        dataRow.put("rowProcessErrorFlag", 0);
        dataRow.put("tid", tid);

        customPutRow(dataRow);

        for (int i = 0; i < rowData.length; i++) {
            Object rowDatum = rowData[i];
            IValueMeta valueMeta = rowMeta.getValueMeta(i);
            dataRow.put(valueMeta.getName(), rowDatum);
        }
        anylineService.insert(getTableName(), dataRow);
        return true;
    }

    protected void customPutRow(DataRow dataRow) {

    }

    @Override
    public int size() {
        if (ObjectUtil.isNull(anylineService)) {
            return 0;
        }
        if (!ready.get()) {
            return 0;
        }

        String whereCondition = getWhereCondition();
        Long count = anylineService.count(getTableName(), StrUtil.nullToDefault(whereCondition, " 1=1 ") + " and rowReadFlag = 0 and tid ='"+tid+"'");
        return count.intValue();
    }


    @Override
    public void clear() {
        ready.set(false);
    }

    public void readFinish(Object dataInput) {

//        if(dataInput instanceof Collection<?> daList){
//
//
//        }

    }

    public void error(Object row, Throwable throwable) {

        if (ArrayUtil.isEmpty(row)) {
            return;
        }
        if (ObjectUtil.isNull(anylineService)) {
            return;
        }
        if (row instanceof Collection<?> collection) {
            for (Object o : collection) {
                Object[] objects = ValueMetaConvert.objToRow(o);
                error(objects, throwable);

            }
        } else {
            Object[] objects = ValueMetaConvert.objToRow(row);
            error(objects, throwable);
        }


    }

    public void error(Object[] row, Throwable throwable) {

        if (ArrayUtil.isEmpty(row)) {
            return;
        }
        if (ObjectUtil.isNull(anylineService)) {
            return;
        }

        Object o = ArrayUtil.get(row, row.length - 1);
        if (!(o instanceof String str)) {
            return;
        }
        if (!StrUtil.startWith(str, "hIndex_")) {
            return;
        }

        str = StrUtil.replace(str, "hIndex_", "");

        DataRow dataRow = new DataRow();
        dataRow.put("ID", str);
//        dataRow.put("rowReadFlag", 0);
        dataRow.put("rowProcessErrorFlag", 1);
        dataRow.put("rowProcessErrorMsg", ExceptionUtil.stacktraceToString(throwable));
        anylineService.update(getTableName(), dataRow);

    }


    public DataSource getDataSource(String date, String processId, String processVersion, String tid, String dbFile, String source, String target) {
        return null;
    }


    public void errorToNoRead() {
        String tableName = getTableName();
        DataRow query = null;

        Table table = new Table(tableName);

//        anylineService.tables(tableName).ex
        boolean exists = anylineService.metadata().exists(table);
        //不存在表，暂时不管
        if (BooleanUtil.isFalse(exists)) {
            return;
        }
        String whereCondition = getWhereCondition();
        synchronized (tableName) {
            query = anylineService.query(tableName, StrUtil.nullToDefault(whereCondition, " 1=1 ") + " and rowProcessErrorFlag = 1 and tid = '"+tid+"'");
            if (ObjectUtil.isNull(query)) {
                //存在未读的也要搭上标记
                DataRow dataRow = anylineService.query(tableName, StrUtil.nullToDefault(whereCondition, " 1=1 ") + " and rowReadFlag = 0 and tid = '"+tid+"'");
                if (ObjectUtil.isNotNull(dataRow)) {
                    ready.set(true);
                    initRowMeta(dataRow);
                }
                return;
            }
            //
            DataRow row = new DataRow();
            row.put("ID", query.getId());
            row.put("rowReadFlag", 0);
            row.put("rowProcessErrorFlag", 0);
            row.set("endTime", DateUtil.date());
            anylineService.update(tableName, row);
            ready.set(true);
            initRowMeta(query);
        }
    }

    public void initRowMeta(DataRow query) {
        if (ObjectUtil.isNotNull(rowMeta)) {
            return;
        }
        synchronized (this) {
            if (ObjectUtil.isNotNull(rowMeta)) {
                return;
            }
            Map<String, Object> map = filterSystemField(query);
            this.rowMeta = ProcessRowDescriptor.objectToRowMeta(map);
        }

    }

    private List<Object> dbDataToRow(DataRow query) {
        List<Object> obj = Lists.newArrayList();
        Map<String, Object> map = filterSystemField(query);
        for (Map.Entry<String, Object> m : map.entrySet()) {
            obj.add(m.getValue());
        }
        //增加一列索引列，用于后续使用
        obj.add("hIndex_" + query.get("id"));
        return obj;
    }

    private Map<String, Object> filterSystemField(DataRow query) {
        Map<String, Object> resultMap = Maps.newHashMap();
        for (Map.Entry<String, Object> sb : query.entrySet()) {
            String key = sb.getKey();
            if (StrUtil.containsAnyIgnoreCase(key, "id", "fromNode", "toNode","tid", "rowReadFlag", "startTime", "endTime"
                    , "rowProcessErrorFlag", "rowProcessErrorMsg")) {
                continue;
            }
            resultMap.put(key, sb.getValue());
//            obj.add(sb.getValue());
        }
        return resultMap;
    }

}
